package org.budo.canal.logposition.zookeeper;

import org.budo.support.java.net.util.JavaNetUtil;
import org.budo.support.lang.util.StringUtil;
import org.springframework.beans.factory.FactoryBean;

import com.alibaba.otter.canal.common.zookeeper.ZkClientx;
import com.alibaba.otter.canal.parse.index.CanalLogPositionManager;
import com.alibaba.otter.canal.parse.index.MemoryLogPositionManager;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

/**
 * @author lmw
 */
@Slf4j
@Getter
@Setter
public class BudoLogPositionManagerFactoryBean implements FactoryBean<CanalLogPositionManager> {
    private ZkClientx zkClientx;

    private String zookeeperUrl;

    @Override
    public CanalLogPositionManager getObject() throws Exception {
        ZkClientx zkClientx = this.zkClientx();
        if (null == zkClientx) {
            log.error("#32 zkClientx is null, return MemoryLogPositionManager, this=" + this);
            return new MemoryLogPositionManager();
        }

        return new BudoPeriodMixedZooKeeperLogPositionManager(zkClientx);
    }

    private ZkClientx zkClientx() {
        if (null != this.getZkClientx()) {
            return this.getZkClientx();
        }

        String zookeeperUrl = this.getZookeeperUrl();
        if (StringUtil.isEmpty(zookeeperUrl)) {
            log.error("#33 init, zookeeperUrl=" + zookeeperUrl + ", this=" + this);

            return null;
        }

        if (!JavaNetUtil.isPortUsing(zookeeperUrl)) {
            throw new IllegalArgumentException("#50 not ok zk, zookeeperUrl=" + zookeeperUrl + ", this=" + this);
        }

        ZkClientx zkClientx = new ZkClientx(zookeeperUrl);

        log.info("#40 init zkClientx=" + zkClientx + ", zookeeperUrl=" + zookeeperUrl + ", this=" + this);

        this.setZkClientx(zkClientx);
        return this.getZkClientx();
    }

    @Override
    public Class<?> getObjectType() {
        return CanalLogPositionManager.class;
    }

    @Override
    public boolean isSingleton() {
        return true;
    }
}
